package com.chencong.online.driver.dwd;

import com.chencong.online.bean.OrderEventBean;
import com.chencong.online.bean.ReceiptEventBean;
import com.chencong.online.env.FlinkEnv;
import com.chencong.online.function.OrderEventMapFunc;
import com.chencong.online.function.PayReceiptProJoinFunc;
import com.chencong.online.function.ReceiptEventMapFunc;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.net.URL;
import java.util.Objects;

/**
 * @program: user-behavior-analysis-online
 * @ClassName DwdPayReceiptJJoinDriver
 * @description:双流join
 * 缺点：不能拿到延迟数据
 * @author: chencong
 * @create: 2022-01-06 16:29
 **/
public class DwdPayReceiptJoinDriver {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = FlinkEnv.FlinkDataStreamRunEnv();
        env.setParallelism(1);
        //1、source
        //1.1 读取支付数据
        URL resource1 = DwdOrderPayTimeOutDriver.class.getResource("/OrderLog.csv");
        DataStreamSource<String> payInputDS = env.readTextFile(resource1.getPath());


        //1.2 读取到账数据
        URL resource2 = DwdOrderPayTimeOutDriver.class.getResource("/ReceiptLog.csv");
        DataStreamSource<String> receiptInputDS = env.readTextFile(resource2.getPath());


//        inputDS.print("初始化数据");

        //2、transform
        //2.1 将支付订单数据数据格式化并提取时间戳
        SingleOutputStreamOperator<OrderEventBean> orderEventBeanDS = payInputDS.map(new OrderEventMapFunc())
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<OrderEventBean>forMonotonousTimestamps() //升序
                        .withTimestampAssigner(new SerializableTimestampAssigner<OrderEventBean>() {
                            @Override
                            public long extractTimestamp(OrderEventBean element, long recordTimestamp) {
                                return element.getTimestamp() * 1000L;
                            }
                        })
                )
                .filter(data -> !Objects.equals(data.getTxId(), "")); //交易Id不能为空，必须是pay事件
//        orderEventBeanDS.print("订单数据局");
        //2.2 将到账订单数据数据格式化并提取时间戳
        SingleOutputStreamOperator<ReceiptEventBean> receiptEventBeanDS = receiptInputDS.map(new ReceiptEventMapFunc())
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<ReceiptEventBean>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<ReceiptEventBean>() {
                            @Override
                            public long extractTimestamp(ReceiptEventBean element, long recordTimestamp) {
                                return element.getTimestamp() * 1000L;
                            }
                        }));
//        receiptEventBeanDS.print("到账数据");

        //双流join
        SingleOutputStreamOperator<Tuple2<OrderEventBean, ReceiptEventBean>> resultDS = orderEventBeanDS.keyBy(OrderEventBean::getTxId)
                .intervalJoin(receiptEventBeanDS.keyBy(ReceiptEventBean::getTxId))
                .between(Time.seconds(-3), Time.seconds(5))
                .process(new PayReceiptProJoinFunc());

        resultDS.print("结果集");
        env.execute("DwdPayReceiptJoinDriver");
    }
}
